Problem

The Position class uses multi-threaded data structures and safe
publication which suggests to users that it's thread safe. The
Position#merge method is not thread-safe due to the check-then-act
logic:

if (!partitionMap.containsKey(partition) || partitionMap.get(partition)
< offset) {
  partitionMap.put(partition, offset);
}

The partitionMap's entry for partition can change between the
conditional check and the put call.

The multi-threaded unit test reproduces the problem and fails ~80% of
the time without the fix.

Solution

Delegate mutation to withComponent which is thread-safe since it
relies on ConcurrentHashMap#compute which uses synchronization
internally to atomically execute the remapping function.